package com.hkbigdata.ontimer;

import com.hkbigdata.entry.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author liuanbo
 * @creat 2023-05-17-17:39
 * @see 2194550857@qq.com
 * ws_1001,1684121131000,3
 * ws_1001,1684121139000,2
 * ws_1001,1684121140000,1
 * ws_1001,1684121142000,4
 * ws_1001,1684121147000,5
 * ws_1001,1684121148000,6
问两个问题？1.数据啥时候触发，2.定时器触发的时间
 * 1684121131000
 * WaterSensor(id=ws_1001, ts=1684121131000, vs=3)
 * 1684121139000
 * WaterSensor(id=ws_1001, ts=1684121139000, vs=2)=》第二条数据为39的时候，触发48，触发定时器的时间为为44
 * 1684121140000
 * WaterSensor(id=ws_1001, ts=1684121140000, vs=1) =>第一条数据为31数据的时候，40触发，触发定时器的时间为36
 * 定时器被触发1684121136000
 * 1684121142000
 * WaterSensor(id=ws_1001, ts=1684121142000, vs=4)
 * 1684121147000
 * WaterSensor(id=ws_1001, ts=1684121147000, vs=5)
 * 1684121148000
 * WaterSensor(id=ws_1001, ts=1684121148000, vs=6)
 * 定时器被触发1684121144000
 */
public class Flink02_EventTime_OnTimer {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取端口数据并转换为JavaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDS = env.socketTextStream("hadoop102", 9999)
                .map(data -> {
                    String[] split = data.split(",");
                    return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));
                });

        //3.策略
        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy.<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
            @Override
            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                return element.getTs();
            }
        });
        //4.分配策略
        SingleOutputStreamOperator<WaterSensor> process = waterSensorDS.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy)
                .keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                        //5.获取事件时间
                        Long timestamp = ctx.timestamp();

                        //6.获取事件时间注册一个5秒的定时器
                        long ts = timestamp + 5000L;
                        System.out.println("注册一个定时器:" + ts);
                        ctx.timerService().registerEventTimeTimer(ts);
                        out.collect(value);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<WaterSensor> out) throws Exception {
                        System.out.println("定时器触发" + timestamp);
                    }
                });

        process.print();

        env.execute();
    }
}
